/**
 * 
 */
package mapreduce4j;

import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.UTFDataFormatException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import org.apache.hadoop.io.WritableComparable;
import org.apache.log4j.Logger;

/**
 * Collects in memory during writing, then sorts and writes to a file on 
 * closing
 * @author tim
 */
public class SortingFileRecordWriter extends RecordWriter {
	protected Logger logger = Logger.getLogger(this.getClass().getName());
	
	// little container object
	class KVP {
		public WritableComparable k;
		public WritableComparable v;
		public KVP(WritableComparable k, WritableComparable v) {
			this.k = k;
			this.v = v;
		}
		
	}
	
	protected List<KVP> data = new ArrayList<KVP>();
	protected File outputFile;

	
	public SortingFileRecordWriter(String s) throws IOException {
		outputFile = new File(s);
		if (!outputFile.exists()) {
			outputFile.createNewFile();
		}
	}
	
	/**
	 * sort it, write it
	 */
	@Override
	public void close() throws IOException, InterruptedException {
		logger.debug("Sorting in memory collection of size["+ data.size() + "]");
		DataOutputStream output;
		Collections.sort(data, new Comparator<KVP>() {
			public int compare(KVP o1, KVP o2) {
				return o1.k.compareTo(o2.k);
			}
		});
		
		output = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(outputFile)));
		for (KVP kvp : data) {
			try {
				kvp.k.write(output);
				kvp.v.write(output);
			} catch (UTFDataFormatException e) {
				// TODO - investigate the error with: encoded string too long: 91120 bytes
				logger.error("Loosing record...", e);
				logger.error("k: " + kvp.k); // some dodgy string happening here
				logger.error("v: " + kvp.v);
			}
		}
		
		output.flush();
		output.close();
			
		logger.debug("Written sorted file: " + outputFile.getAbsolutePath());
	}

	/**
	 * writes to the file 
	 */
	@Override
	public void write(WritableComparable key, WritableComparable value) throws IOException,InterruptedException {
		data.add(new KVP(key, value));
	}
}
